package com.example.realtimeeb.dim

import com.example.realtimeeb.model.Area
import org.apache.flink.streaming.api.{CheckpointingMode, TimeCharacteristic}
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.api.scala._
import org.apache.flink.table.api.Table
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
import org.apache.flink.types.Row

/**
 * 首先从HBase中读取lagou_area的数据，然后转化成宽表，最后sink到dim_area宽表中
 */
object TransformAreaToWideTable {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(env)
    env.enableCheckpointing(5000)
//    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
//    env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)

    val rawData: DataStream[(String, String)] = env.addSource(new HbaseReader)

    rawData.print()

    //获取几个必要的字段id name pid
    val mapped: DataStream[Area] = rawData.map(x => {
      val id: Int = x._1.toInt
      val datas: Array[String] = x._2.split("-")
      val name: String = datas(5).trim
      val pid: Int = datas(6).trim.toInt

      Area(id, name, pid)
    })

    mapped.print()

    //临时表
    tableEnv.createTemporaryView("lagou_area", mapped)

    //sql -- 生成 区、市、省三级的明细宽表
    val sql : String =
      """
        |select a.id as areaid,a.name as aname,a.pid as cid,b.name as city, c.id as proid,c.name as province
        |from lagou_area as a
        |inner join lagou_area as b on a.pid = b.id
        |inner join lagou_area as c on b.pid = c.id""".stripMargin

    val areaTable: Table = tableEnv.sqlQuery(sql)

    areaTable.printSchema()

    val resultStream: DataStream[String] = tableEnv.toRetractStream[Row](areaTable).map(x => {
      val row: Row = x._2
      val areaId: String = row.getField(0).toString
      val aname: String = row.getField(1).toString
      val cid: String = row.getField(2).toString
      val city: String = row.getField(3).toString
      val proid: String = row.getField(4).toString
      val province: String = row.getField(5).toString

      areaId + "," + aname + "," + cid + "," + city + "," + proid + "," + province
    })

    resultStream.print()

    resultStream.addSink(new HbaseWriterSink)
    env.execute()
  }

}
